package com.uptown.matrix.core.producer.instance;

import java.util.Map;
import java.util.Properties;

import javax.annotation.Resource;

import com.uptown.matrix.core.producer.ProducerService;
import lombok.SneakyThrows;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import com.uptown.matrix.common.result.WebResult;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService extends ProducerService {

    @Resource
    Map<String, KafkaConsumer<String, String>> kafkaConsumerMap;

    /**
     * 发送端自行处理异常
     */
    @SneakyThrows
    @Override
    public WebResult producerSend(String bootServer, String group, String topic, String message) {
        // FIXME: 白名单校验
        super.producerSend(bootServer, group, topic, message);

        Properties props = new Properties();
        props.put("bootstrap.servers", bootServer);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 创建生产者
        Producer<String, String> producer = new KafkaProducer<>(props);
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);
        RecordMetadata recordMetadata = producer.send(record).get();
        producer.close();
        return WebResult.of(recordMetadata.offset());
    }

    @Override
    public WebResult topicList() {
        return WebResult.of(kafkaConsumerMap.keySet());
    }
}
